package com.sec.etech.redis.stream.consumer;

import com.sec.etech.redis.stream.RedisStreamUtil;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.PendingMessages;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service
public class TestConsumeService {
    @Resource
    private RedisStreamUtil redisStreamUtil;

    //消息消息
    public void consumeMessages(String streamKey, String group, String consumerName) {

        redisStreamUtil.initGroup(streamKey, group);
        while (true) {
            List<MapRecord<String, Object, Object>> msgList = redisStreamUtil.readLastConsumed(streamKey, group, consumerName);
            if (msgList == null || msgList.size() == 0) {
                continue;
            }
            try {
                this.handleMessage(msgList);
            } catch (Exception e) {
                e.printStackTrace();
                while (true) {
                    List<MapRecord<String, Object, Object>> errorMsgList = redisStreamUtil.readById(streamKey, group, consumerName, "0");
                    if (errorMsgList == null || errorMsgList.size() == 0) {
                        continue;
                    }
                    try {
                        this.handleMessage(errorMsgList);
                    } catch (Exception e2) {
                        System.out.println("handleErrorMessage报错");
                        e2.printStackTrace();
                    }
                }
            }
        }
    }

    private void handleMessage(List<MapRecord<String, Object, Object>> msgList) {
        for (MapRecord<String, Object, Object> record : msgList) {
            System.out.println("%%%%%%%%%%%%%%TestConsumeService.handleMessage：" + record);
        }
        //业务处理成功后确认消费
    }

    /*private void handleErrorMessage(String streamKey, String group, String consumerName) {
        List<MapRecord<String, Object, Object>> msgList = redisStreamUtil.readAllPEL(streamKey, group, consumerName);
        this.handleMessage(msgList);
    }*/

    public void pendingList() throws Exception {
        //方法1：先获取所有ID，再根据ID获取消息
        //PendingMessages msgs = redisStreamUtil.readPendingListId("test_stream", "group-b", "consumer-bb");
        PendingMessages msgs = redisStreamUtil.readPendingListId("nList-bpm-create2", "group-bpm-create", "consumer-a");
        msgs.forEach((msg)->{
            //System.out.println("方法1：" + msg.toString());
            System.out.println("方法1：" + msg.getConsumerName() + ",id=" + msg.getId().toString());
            List<MapRecord<String, Object, Object>> list = redisStreamUtil.readById(
                    "nList-bpm-create2", "group-bpm-create", "consumer-a", msg.getId().getSequence().toString());
            for(MapRecord<String, Object, Object> record:list){
                System.out.println("方法1 %% ：" + record.getValue());
            }
        });
        //方法2：直接获取指定数量的消息
        //List<MapRecord<String, Object, Object>> list =
                //redisStreamUtil.readPendingList("test_stream", "group-b", "consumer-bb");
        List<MapRecord<String, Object, Object>> list =
            redisStreamUtil.readPendingList("nList-bpm-create2", "group-bpm-create", "consumer-a");
        for(MapRecord<String, Object, Object> record:list){
            System.out.println("方法2：" + record.getValue());
        }
    }
    //处理pending-list:
    //@PostConstruct
    private void handlerPendingList(String streamKey) {
        //streamKey = "test_stream";
        while (true){
            //死循环 一直执行消息队列里面的任务
            try {
                //读取未被确认的消息
                List<MapRecord<String, Object, Object>> recordList = redisStreamUtil.readPendingList(
                        streamKey, "group-a", "consumer-a");
                if (recordList==null||recordList.isEmpty()){
                    //没有获取到消息,直接退出循环
                    break;
                }
                MapRecord<String, Object, Object> record = recordList.get(0);
                //获取消息,构建选课信息表
                Map<Object, Object> mp = record.getValue();
                String cno = (String) mp.get("cno");
                String uno = (String) mp.get("uno");
                //buildSckillCourse(cno,uno);
                //最后需要确认这个消息已经被消费
                //redisStreamUtil.ack(streamKey, "group-a", record.getId().getValue());
            }catch (Exception e){
                //log.debug("pedding-list消费异常..",e);
                try {
                    Thread.sleep(200);
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
            }
        }
    }

    //生产消息
    public void addMessages(String streamKey, String key, Object value) {
        Map<String, Object> content = new HashMap<>();
        content.put(key, value);
        redisStreamUtil.add(streamKey, 123L, content);
    }
}
